工具调用机制概述#

工具调用（Tool Calling）是Agentic AI系统的核心能力之一，它允许AI模型通过调用外部工具来扩展其能力，执行实际的操作，而不仅仅是生成文本。

工具调用的基本概念#

1. 什么是工具调用#

工具调用是指AI模型根据用户的请求，识别需要使用的工具，生成工具调用参数，执行工具调用，并将结果整合到最终响应中的过程。

工具调用的特点:

• 结构化输出: 模型输出结构化的工具调用指令
• 参数验证: 验证工具调用参数的有效性
• 结果整合: 将工具执行结果整合到响应中
• 错误处理: 处理工具调用失败的情况

2. 工具调用流程#

用户请求 ↓ 意图识别 ↓ 工具选择 ↓ 参数生成 ↓ 工具执行 ↓ 结果处理 ↓ 响应生成

工具定义与注册#

1. 工具定义#

python
复制
python

# 示例：工具定义

用户请求：
"定义一个文件读取工具"

Claude Code 生成的代码：

```python
```python

from typing import Dict, Any, Optional
import json
from dataclasses import dataclass, asdict
from enum import Enum

class ToolType(Enum):
    """工具类型"""
    FILE_OPERATION = "file_operation"
    CODE_EXECUTION = "code_execution"
    API_CALL = "api_call"
    SEARCH = "search"
    CALCULATION = "calculation"

@dataclass
class ToolParameter:
    """工具参数"""
    name: str
    type: str
    description: str
    required: bool
    default: Optional[Any] = None

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    type: ToolType
    parameters: list[ToolParameter]
    function: callable

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            'name': self.name,
            'description': self.description,
            'type': self.type.value,
            'parameters': [asdict(param) for param in self.parameters]
        }

# 文件读取工具
def read_file(file_path: str) -> str:
    """读取文件内容"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found: {file_path}"
    except Exception as e:
        return f"Error: {str(e)}"

read_file_tool = ToolDefinition(
    name="read_file",
    description="读取指定文件的内容",
    type=ToolType.FILE_OPERATION,
    parameters=[
        ToolParameter(
            name="file_path",
            type="string",
            description="要读取的文件路径",
            required=True
        )
    ],
    function=read_file
)

# 文件写入工具
def write_file(file_path: str, content: str) -> str:
    """写入内容到文件"""
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Success: File written to {file_path}"
    except Exception as e:
        return f"Error: {str(e)}"

write_file_tool = ToolDefinition(
    name="write_file",
    description="将内容写入到指定文件",
    type=ToolType.FILE_OPERATION,
    parameters=[
        ToolParameter(
            name="file_path",
            type="string",
            description="要写入的文件路径",
            required=True
        ),
        ToolParameter(
            name="content",
            type="string",
            description="要写入的内容",
            required=True
        )
    ],
    function=write_file
)

# 代码执行工具
def execute_code(code: str, language: str = "python") -> str:
    """执行代码"""
    try:
        if language == "python":
            import subprocess
            result = subprocess.run(
                ['python', '-c', code],
                capture_output=True,
                text=True,
                timeout=10
            )
            if result.returncode == 0:
                return result.stdout
            else:
                return f"Error: {result.stderr}"
        else:
            return f"Error: Unsupported language: {language}"
    except subprocess.TimeoutExpired:
        return "Error: Code execution timeout"
    except Exception as e:
        return f"Error: {str(e)}"

execute_code_tool = ToolDefinition(
    name="execute_code",
    description="执行代码并返回结果",
    type=ToolType.CODE_EXECUTION,
    parameters=[
        ToolParameter(
            name="code",
            type="string",
            description="要执行的代码",
            required=True
        ),
        ToolParameter(
            name="language",
            type="string",
            description="编程语言",
            required=False,
            default="python"
        )
    ],
    function=execute_code
)

### 2. 工具注册

```python

# 示例：工具注册

用户请求：
"实现一个工具注册系统"

Claude Code 生成的代码：

```python
```python

from typing import Dict, List, Optional

class ToolRegistry:
    """工具注册表"""

    def __init__(self):
        self.tools: Dict[str, ToolDefinition] = {}
        self.categories: Dict[str, List[str]] = {}

    def register(self, tool: ToolDefinition, category: Optional[str] = None):
        """注册工具"""
        self.tools[tool.name] = tool

        if category:
            if category not in self.categories:
                self.categories[category] = []
            self.categories[category].append(tool.name)

    def unregister(self, tool_name: str):
        """注销工具"""
        if tool_name in self.tools:
            del self.tools[tool_name]

            for category, tools in self.categories.items():
                if tool_name in tools:
                    tools.remove(tool_name)

    def get_tool(self, tool_name: str) -> Optional[ToolDefinition]:
        """获取工具"""
        return self.tools.get(tool_name)

    def list_tools(self, category: Optional[str] = None) -> List[ToolDefinition]:
        """列出工具"""
        if category:
            tool_names = self.categories.get(category, [])
            return [self.tools[name] for name in tool_names]
        else:
            return list(self.tools.values())

    def search_tools(self, keyword: str) -> List[ToolDefinition]:
        """搜索工具"""
        results = []
        keyword = keyword.lower()

        for tool in self.tools.values():
            if (keyword in tool.name.lower() or
                keyword in tool.description.lower()):
                results.append(tool)

        return results

    def get_tool_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
        """获取工具schema"""
        tool = self.get_tool(tool_name)
        if tool:
            return tool.to_dict()
        return None

    def get_all_schemas(self) -> Dict[str, Dict[str, Any]]:
        """获取所有工具schema"""
        return {
            name: tool.to_dict()
            for name, tool in self.tools.items()
        }

# 使用示例
registry = ToolRegistry()

# 注册工具
registry.register(read_file_tool, category="file_operations")

registry.register(write_file_tool, category="file_operations") registry.register(execute_code_tool, category="code_execution")

列出所有工具

print("All tools:") for tool in registry.list_tools(): print(f" - {tool.name}: {tool.description}")

列出特定类别的工具

print("\nFile operation tools:") for tool in registry.list_tools(category="file_operations"): print(f" - {tool.name}: {tool.description}")

搜索工具

print("\nSearch results for 'file':") for tool in registry.search_tools("file"): print(f" - {tool.name}: {tool.description}")

获取工具schema

print("\nTool schema for read_file:") print(json.dumps(registry.get_tool_schema("read_file"), indent=2))

工具调用执行#

1. 工具调用解析#

示例：工具调用解析

用户请求： "实现工具调用解析器" Claude Code 生成的代码：

python
复制
`python

import re
from typing import Dict, Any, List, Optional
import json

class ToolCallParser:
"""工具调用解析器"""

def __init__(self, registry: ToolRegistry):
self.registry = registry

def parse_tool_calls(self, text: str) -> List[Dict[str, Any]]:
"""解析工具调用"""
tool_calls = []

 # 匹配工具调用模式
pattern = r'<tool_call>\s*<name>(.*?)</name>\s*<parameters>(.*?)</parameters>\s*</tool_call>'
matches = re.finditer(pattern, text, re.DOTALL)

for match in matches:
tool_name = match.group(1).strip()
parameters_str = match.group(2).strip()

 # 解析参数
try:
parameters = json.loads(parameters_str)
except json.JSONDecodeError:
parameters = self._parse_parameters(parameters_str)

tool_calls.append({
'tool': tool_name,
'parameters': parameters
})

return tool_calls

def _parse_parameters(self, parameters_str: str) -> Dict[str, Any]:
"""解析参数字符串"""
parameters = {}

 # 匹配参数
param_pattern = r'<(\w+)>(.*?)</\1>'
matches = re.findall(param_pattern, parameters_str, re.DOTALL)

for name, value in matches:
parameters[name] = value.strip()

return parameters

def validate_tool_call(self, tool_call: Dict[str, Any]) -> tuple[bool, Optional[str]]:
"""验证工具调用"""
tool_name = tool_call['tool']
parameters = tool_call['parameters']

 # 检查工具是否存在
tool = self.registry.get_tool(tool_name)
if not tool:
return False, f"Tool not found: {tool_name}"

 # 检查必需参数
for param in tool.parameters:
if param.required and param.name not in parameters:
return False, f"Missing required parameter: {param.name}"

 # 检查参数类型
for param in tool.parameters:
if param.name in parameters:
value = parameters[param.name]
if not self._check_parameter_type(value, param.type):
return False, f"Invalid type for parameter {param.name}: expected {param.type}"

return True, None

def _check_parameter_type(self, value: Any, expected_type: str) -> bool:
"""检查参数类型"""
type_mapping = {
'string': str,
'integer': int,
'float': float,
'boolean': bool,
'array': list,
'object': dict
}

expected_python_type = type_mapping.get(expected_type)
if expected_python_type:
return isinstance(value, expected_python_type)

return True

# 使用示例
parser = ToolCallParser(registry)

# 解析工具调用
text = """
I'll read the file for you.

<tool_call>
<name>read_file</name>
<parameters>{"file_path": "/path/to/file.txt"}</parameters>
</tool_call>
"""

tool_calls = parser.parse_tool_calls(text)
print("Parsed tool calls:")
for tool_call in tool_calls:
print(f"  Tool: {tool_call['tool']}")
print(f"  Parameters: {tool_call['parameters']}")

 # 验证工具调用
is_valid, error = parser.validate_tool_call(tool_call)
if is_valid:
print(f"  Status: Valid")
else:
print(f"  Status: Invalid - {error}")

```### 2. 工具调用执行

```
python

# 示例：工具调用执行

用户请求：
"实现工具调用执行器"

Claude Code 生成的代码：

from typing import Dict, Any, List
import asyncio
from concurrent.futures import ThreadPoolExecutor

class ToolExecutor:
"""工具执行器"""

def __init__(self, registry: ToolRegistry):
self.registry = registry
self.executor = ThreadPoolExecutor(max_workers=4)

async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""执行工具调用"""
tool_name = tool_call['tool']
parameters = tool_call['parameters']

 # 获取工具
tool = self.registry.get_tool(tool_name)
if not tool:
return {
'success': False,
'error': f"Tool not found: {tool_name}",
'tool': tool_name
}

try:
 # 执行工具
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
self.executor,
tool.function,
 **parameters
)

return {
'success': True,
'result': result,
'tool': tool_name
}

except Exception as e:
return {
'success': False,
'error': str(e),
'tool': tool_name
}

async def execute_tool_calls(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""执行多个工具调用"""
tasks = [
self.execute_tool_call(tool_call)
for tool_call in tool_calls
]

results = await asyncio.gather(*tasks, return_exceptions=True)

return results

def execute_tool_call_sync(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
"""同步执行工具调用"""
tool_name = tool_call['tool']
parameters = tool_call['parameters']

 # 获取工具
tool = self.registry.get_tool(tool_name)
if not tool:
return {
'success': False,
'error': f"Tool not found: {tool_name}",
'tool': tool_name
}

try:
 # 执行工具
result = tool.function(**parameters)

return {
'success': True,
'result': result,
'tool': tool_name
}

except Exception as e:
return {
'success': False,
'error': str(e),
'tool': tool_name
}

# 使用示例
executor = ToolExecutor(registry)

# 执行工具调用
tool_calls = [
{
'tool': 'read_file',
'parameters': {'file_path': '/path/to/file.txt'}
},
{
'tool': 'execute_code',
'parameters': {'code': 'print("Hello, World!")', 'language': 'python'}
}
]

# 异步执行
async def main():
results = await executor.execute_tool_calls(tool_calls)

print("Execution results:")
for result in results:
if result['success']:
print(f"  {result['tool']}: Success")
print(f"    Result: {result['result']}")
else:
print(f"  {result['tool']}: Failed")
print(f"    Error: {result['error']}")

# 同步执行
def main_sync():
for tool_call in tool_calls:
result = executor.execute_tool_call_sync(tool_call)

if result['success']:
print(f"{result['tool']}: Success")
print(f"  Result: {result['result']}")
else:
print(f"{result['tool']}: Failed")
print(f"  Error: {result['error']}")

`> >
## 工具调用优化
### 1. 缓存机制
```python
python

# 示例：工具调用缓存

用户请求：
"实现工具调用缓存"

Claude Code 生成的代码：

```python
```python

from typing import Dict, Any, Optional
import hashlib
import json
from datetime import datetime, timedelta

class ToolCallCache:
    """工具调用缓存"""

    def __init__(self, ttl: int = 3600):
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.ttl = ttl

    def _generate_key(self, tool_name: str, parameters: Dict[str, Any]) -> str:
        """生成缓存键"""
        key_data = {
            'tool': tool_name,
            'parameters': parameters
        }
        key_str = json.dumps(key_data, sort_keys=True)
        return hashlib.md5(key_str.encode()).hexdigest()

    def get(self, tool_name: str, parameters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """获取缓存"""
        key = self._generate_key(tool_name, parameters)

        if key in self.cache:
            cached = self.cache[key]

            # 检查是否过期
            if datetime.utcnow() - cached['timestamp'] < timedelta(seconds=self.ttl):
                return cached['result']
            else:
                del self.cache[key]

        return None

    def set(self, tool_name: str, parameters: Dict[str, Any], result: Dict[str, Any]):
        """设置缓存"""
        key = self._generate_key(tool_name, parameters)

        self.cache[key] = {
            'result': result,
            'timestamp': datetime.utcnow()
        }

    def clear(self):
        """清空缓存"""
        self.cache.clear()

    def cleanup(self):
        """清理过期缓存"""
        current_time = datetime.utcnow()
        expired_keys = [
            key for key, cached in self.cache.items()
            if current_time - cached['timestamp'] >= timedelta(seconds=self.ttl)
        ]

        for key in expired_keys:
            del self.cache[key]

class CachedToolExecutor(ToolExecutor):
    """带缓存的工具执行器"""

    def __init__(self, registry: ToolRegistry, cache: ToolCallCache):
        super().__init__(registry)
        self.cache = cache

    async def execute_tool_call(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
        """执行工具调用（带缓存）"""
        tool_name = tool_call['tool']
        parameters = tool_call['parameters']

        # 检查缓存
        cached_result = self.cache.get(tool_name, parameters)
        if cached_result:
            return cached_result

        # 执行工具调用
        result = await super().execute_tool_call(tool_call)

        # 缓存结果
        if result['success']:
            self.cache.set(tool_name, parameters, result)

        return result

### 2. 批量执行

```python

# 示例：批量工具调用

用户请求：
"实现批量工具调用"

Claude Code 生成的代码：

```python
```python

from typing import Dict, Any, List
import asyncio

class BatchToolExecutor(ToolExecutor):
    """批量工具执行器"""

    async def execute_batch(self, batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """批量执行工具调用"""
        # 按工具类型分组
        tool_groups = {}
        for tool_call in batch:
            tool_name = tool_call['tool']
            if tool_name not in tool_groups:
                tool_groups[tool_name] = []
            tool_groups[tool_name].append(tool_call)

        # 并行执行不同工具的调用
        tasks = [
            self._execute_tool_group(tool_name, tool_calls)
            for tool_name, tool_calls in tool_groups.items()
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 合并结果
        all_results = []
        for result_list in results:
            if isinstance(result_list, list):
                all_results.extend(result_list)

        return all_results

    async def _execute_tool_group(self, tool_name: str, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """执行同一工具的多个调用"""
        tool = self.registry.get_tool(tool_name)
        if not tool:
            return [
                {
                    'success': False,
                    'error': f"Tool not found: {tool_name}",
                    'tool': tool_name
                }
                for _ in tool_calls
            ]

        results = []
        for tool_call in tool_calls:
            result = await super().execute_tool_call(tool_call)
            results.append(result)

        return results

```

## 总结

工具调用机制包括：

1. **工具调用的基本概念**: 什么是工具调用、工具调用流程
2. **工具定义与注册**: 工具定义、工具注册
3. **工具调用执行**: 工具调用解析、工具调用执行
4. **工具调用优化**: 缓存机制、批量执行

通过工具调用机制，Claude Code可以执行各种实际操作，大大扩展了其能力。

在下一节中，我们将探讨自主规划算法。

```
```